﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Linq.Expressions;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using Microsoft.Xrm.Client;
using Sustainalytics.DynamicsCRMGateway;
using System.Threading;

namespace Sustainalytics.Universes.Synchronization
{
    public class Crm2MemoryDataMover<T> where T : CrmEntity
    {
        private TransformBlock<int, T[]> _producer;
        private ActionBlock<T[]> _consumer;
        private ICRMSimpleReader<T> _reader;
        private readonly int _maxDegreeOfParallelism;
        private ConcurrentBag<T[]> _bag = new ConcurrentBag<T[]>();

        public Crm2MemoryDataMover(int maxDegreeOfParallelism, ICRMSimpleReader<T> reader)
        {
            _maxDegreeOfParallelism = maxDegreeOfParallelism;
            _reader = reader;
            SetupFlow();
        }

        public Crm2MemoryDataMover(int maxDegreeOfParallelism, string connectionString, int pageSize, IEnumerable<Expression<Func<T, object>>> projections, IEnumerable<WhereClause<T, object>> filters)
        {
            _bag = new ConcurrentBag<T[]>();
            _maxDegreeOfParallelism = maxDegreeOfParallelism;
            _reader = new CRMSimpleReader<T>(connectionString, pageSize, projections, filters);
            SetupFlow();

        }

        public Task<IEnumerable<T>> Run()
        {
            return (new TaskFactory<IEnumerable<T>>()).StartNew(() =>
            {
                using (_reader)
                {
                    using (_producer.LinkTo(_consumer, new DataflowLinkOptions() { PropagateCompletion = true }))
                    {
                        _producer.Post(1);
                        _consumer.Completion.Wait();
                    }
                    return _bag.SelectMany(x => x);
                }
            }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
        }

        private void SetupFlow()
        {
            _producer = new TransformBlock<int, T[]>(x => _reader.ReadNextPage(), new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = 1});
            _consumer = new ActionBlock<T[]>(x => Consume(x), new ExecutionDataflowBlockOptions() { MaxDegreeOfParallelism = _maxDegreeOfParallelism });
        }

        private void Consume(T[] input)
        {
            if (input != null && input.Any())
            {
                _bag.Add(input);
                _producer.Post(1);
            }
            else
            {
                _producer.Complete();
            }
        }
    }

}
